package com.example.canal.strategy.service;

import com.example.canal.common.bean.CanalBean;
import com.example.canal.common.constant.MysqlDBConstant;
import com.example.canal.common.util.DbChangeExecutor;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;


@Configuration
@ConfigurationProperties(prefix = "canal.database.tablename")
@EnableConfigurationProperties(AbstractDBChangeService.class)
@Data
public abstract class AbstractDBChangeService {

    private static final Logger log = LoggerFactory.getLogger(AbstractDBChangeService.class);

    private List<String> list = new ArrayList<>();


    public void process(CanalBean canalBean) {
        String type = canalBean.getType();
        if (list.contains(canalBean.getTable()) && type != null) {
            Executor executor = DbChangeExecutor.getExecutorByTableName(canalBean);
            List<String> data = canalBean.getData();
            if (MysqlDBConstant.TYPE_INSERT.equals(type)) {
                execIfExecutor(executor,() -> onInsert(data));
            } else if (MysqlDBConstant.TYPE_DELETE.equals(type)) {
                execIfExecutor(executor,() -> onDelete(data));
            } else if (MysqlDBConstant.TYPE_UPDATE.equals(type)) {
                execIfExecutor(executor,() -> onUpdate(data,canalBean.getOld()));
            }
        } else {
            log.warn("tableName:{} is not avaiable to  process  type:{}",canalBean.getTable(),type);
            return;
        }

    }


    protected abstract void onInsert(List<String> data);

    protected abstract void onDelete(List<String> data);

    protected abstract void onUpdate(List<String> data,List<String> old);


    private void execIfExecutor(Executor executor,Runnable runnable) {
        if (executor == null) {
            runnable.run();
        } else {
            executor.execute(runnable);
        }
    }

    public void setList(List<String> list) {
        List<String> stringList = list
                .stream()
                .map(String::toLowerCase)
                .collect(Collectors.toList());

        this.list = stringList;
    }
}
